package com.tian.concurrent;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executors;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


public class ThreadPoolExecutorTest {
//	final BlockingQueue<Runnable> queue = new SynchronousQueue<Runnable>();
	final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(10);
	final ThreadPoolExecutor excutor = new ThreadPoolExecutor(10, 600, 30, TimeUnit.SECONDS, queue, Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
	final AtomicInteger completedTask = new AtomicInteger(0);
	final AtomicInteger rejectedTask = new AtomicInteger(0);
	
	private Lock lock = new ReentrantLock();
	
	static long beginTime;
	final int count = 1000;
	
	public static void main(String[] args) {
		beginTime = System.currentTimeMillis();
		new ThreadPoolExecutorTest().start();
	}
	
	public void start(){
		//启动1000个线程
		CountDownLatch latch = new CountDownLatch(count);//CountDownLatch 在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。
		CyclicBarrier barrier = new CyclicBarrier(count);//CyclicBarrier是一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point) CyclicBarrier数的是调用了CyclicBarrier.await()进入等待的线程数，当线程数达到了CyclicBarrier初始时规定的数目时，所有进入等待状态的线程被唤醒并继续。
		
		for(int i = 0;i < count;i++){
			new Thread(new TestThread(latch, barrier)).start();
		}
		try {
			latch.await(); //await方法，调用此方法会一直阻塞当前线程，直到计时器的值为0
			excutor.shutdownNow();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		
	}
	
	class TestThread implements Runnable{
		private CountDownLatch latch;
		private CyclicBarrier barrier; 
		public TestThread(CountDownLatch latch,CyclicBarrier barrier) {
			this.latch = latch;
			this.barrier = barrier;
		}
		
		public void run() {
			try {
				barrier.await(); //在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (BrokenBarrierException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			excutor.execute(new Task(latch));
		}
	}
	
	class Task implements Runnable{
		
		private CountDownLatch latch;
		public Task(CountDownLatch latch) {
			this.latch = latch;
		}
		public void run() {
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			lock.lock();
			System.out.println("执行的任务数为："+completedTask.incrementAndGet());
			System.out.println("任务耗时为："+(System.currentTimeMillis()-beginTime)+" ms");				
			lock.unlock();
			
			latch.countDown(); //countDown方法，当前线程调用此方法，则计数减一  
		}
	}
}



